package com.epoint.flinkdemo.window;

import com.epoint.flinkdemo.entity.SenSorReading;
import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;

/**
 * @author liufl
 * @version [版本号, 21-4-13]
 */
public class WatermarkSideOutputTest
{
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        DataStreamSource<String> socketStream = env.socketTextStream("localhost", 7777);

        DataStream<SenSorReading> mapStream = socketStream.map(new MapFunction<String, SenSorReading>()
        {
            @Override
            public SenSorReading map(String value) throws Exception {
                String[] str = value.split(",");
                // 时间戳自己造
                return new SenSorReading(str[0], Long.parseLong(str[1]), Double.parseDouble(str[2]));
            }
        });

        // 提取并设置event-time
        // 延迟2秒
        SingleOutputStreamOperator<SenSorReading> watermarks = mapStream
                .assignTimestampsAndWatermarks(
                        new BoundedOutOfOrdernessTimestampExtractor<SenSorReading>(Time.seconds(2L))
                        {
                            @Override
                            public long extractTimestamp(SenSorReading element) {
                                return element.getTimestamp();
                            }
                        });

        OutputTag<SenSorReading> delaySensor = new OutputTag<SenSorReading>("delaySensor"){};
        // 计算5秒之内的平均温度
        SingleOutputStreamOperator<Tuple4<String, String, String, Double>> applyStream = watermarks.keyBy("id")
                .timeWindow(Time.seconds(5L))
                .sideOutputLateData(delaySensor)
                .apply(new WindowFunction<SenSorReading, Tuple4<String, String, String, Double>, Tuple, TimeWindow>()
                {

                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<SenSorReading> input,
                            Collector<Tuple4<String, String, String, Double>> out) throws Exception {
                        // 因为已经按照id进行了分区
                        List<SenSorReading> list = IteratorUtils.toList(input.iterator());
                        Double sum = 0.0;
                        for (SenSorReading senSorReader : list) {
                            sum = sum + senSorReader.getTemp();
                        }

                        out.collect(new Tuple4<>(tuple.getField(0).toString(), DateFormatUtils
                                .format(window.getStart(), "yyyy-MM-dd HH:mm:ss"),
                                DateFormatUtils.format(window.getEnd(), "yyyy-MM-dd HH:mm:ss"), sum / list.size()));
                    }
                });

        // 获取侧输出流
        applyStream.getSideOutput(delaySensor).print();

        applyStream.print();

        env.execute();
    }
}
